package bigbeard.tools.net.client.impl;

import bigbeard.tools.net.api.StateChangeCallback;
import bigbeard.tools.net.client.BaseClientImpl;
import bigbeard.tools.net.utils.Utils;
import bigbeard.tools.thread.pools.ThreadPoolFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author bigbeard
 */
public class TcpClientImpl extends BaseClientImpl {

    private static final String CLIENT_KEY = "client";
    private static final String THREAD_NAME_PREFIX = "tcp_client-thread-data-reader-%s";
    private static final int SERVER_CLOSE = -1;
    private static final int READ_BUFF_SIZE = 4096;
    private static final int CONNECT_TIMEOUT = 5000;

    private final Log logger = LogFactory.getLog(TcpClientImpl.class);

    private ExecutorService executorService;

    private String hostIp = "";
    private int hostPort;
    /**
     * 信道选择器
     */
    private Selector selector = null;
    /**
     * 与服务器通信的信道
     */
    private SocketChannel socketChannel = null;

    private volatile boolean isConnected = false;

    @Override
    public synchronized void connect(String ip, int port, String other) throws Exception {
        if ("".equals(ip) || null == ip) {
            throw new Exception("ip or host must not null");
        }
        if (port <= 0) {
            throw new Exception("端口错误");
        }
        boolean b = hostPort == port && hostIp.equals(ip);
        if (b && isConnected) {
            logger.info("连接已经建立，无需重复连接");
            return;
        }
        hostIp = ip;
        hostPort = port;
        this.disconnect();

        final InetSocketAddress inetSocketAddress = new InetSocketAddress(ip, port);
        try {
            selector = Selector.open();
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);

            if (!socketChannel.connect(inetSocketAddress)) {
                // Register for connect event and wait with timeout
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
                long startTime = System.currentTimeMillis();
                while (true) {
                    if (selector.select(CONNECT_TIMEOUT) > 0) {
                        Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
                        while (keys.hasNext()) {
                            SelectionKey key = keys.next();
                            keys.remove();
                            if (key.isConnectable()) {
                                if (socketChannel.finishConnect()) {
                                    setupConnection();
                                    return;
                                }
                            }
                        }
                    }
                    if (System.currentTimeMillis() - startTime >= CONNECT_TIMEOUT) {
                        throw new IOException("Connection timed out");
                    }
                }
            } else {
                setupConnection();
            }
        } catch (IOException ex) {
            isConnected = false;
            dispose();
            //notifyState(StateChangeCallback.CLIENT_ERROR, "连接失败，错误信息：" + ex.getMessage());
            throw ex;
        }
    }

    private void setupConnection() throws IOException {
        socketChannel.register(selector, SelectionKey.OP_READ);
        isConnected = true;
        if (executorService != null) {
            executorService.shutdownNow();
        }
        executorService = ThreadPoolFactory.createSingleThreadPool(THREAD_NAME_PREFIX);
        executorService.execute(this::readData);
        notifyState(StateChangeCallback.CLIENT_OPENED, "Connected successfully");
    }

    /**
     * 发送数据
     *
     * @param data 待发数据
     * @return 发送长度, 小于0失败
     * @throws IOException 链接异常
     */
    @Override
    public int send(byte[] data) throws Exception {
        if (!isConnected  || socketChannel == null|| !socketChannel.isConnected()) {
            throw new IllegalStateException("连接已经关闭或者未连接");
        }
        return Utils.write(socketChannel, data);
    }

    @Override
    public void send(String msg) throws Exception {
        this.send(msg.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public synchronized void disconnect() {
        isConnected = false;
        if (selector != null) {
            selector.wakeup();
        }
        dispose();
        if (executorService != null) {
            executorService.shutdownNow();
            try {
                if (!executorService.awaitTermination(1, TimeUnit.SECONDS)) {
                    logger.warn("Executor service did not terminate promptly");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            executorService = null;
        }
    }

    @Override
    protected String getClientKey() {
        return CLIENT_KEY;
    }

    private void dispose() {
        try {
            if (socketChannel != null) {
                socketChannel.close();
            }
        } catch (IOException e) {
            logger.debug("Error closing socket channel", e);
        } finally {
            socketChannel = null;
        }

        try {
            if (selector != null) {
                selector.close();
            }
        } catch (IOException e) {
            logger.debug("Error closing selector", e);
        } finally {
            selector = null;
        }
    }
    private void readData() {
        try {
            while (isConnected) {
                if (selector.select() > 0) {
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext() && isConnected) {
                        SelectionKey sk = iterator.next();
                        iterator.remove();

                        if (sk.isValid() && sk.isReadable()) {
                            readFromChannel((SocketChannel) sk.channel());
                        }
                    }
                }
            }
        } catch (IOException ex) {
            logger.error("Read error", ex);
            notifyState(StateChangeCallback.CLIENT_ERROR, "Read error: " + ex.getMessage());
        } finally {
            disconnect();
        }
    }

    private void readFromChannel(SocketChannel channel) {
        ByteBuffer buffer = ByteBuffer.allocate(READ_BUFF_SIZE);
        try {
            int bytesRead;
            while ((bytesRead = channel.read(buffer)) > 0) {
                buffer.flip();
                byte[] data = new byte[buffer.remaining()];
                buffer.get(data);
                notifyDataArrival(data);
                buffer.clear();
            }

            if (bytesRead == SERVER_CLOSE) {
                notifyState(StateChangeCallback.CLIENT_CLOSE, "Server closed connection");
                disconnect();
            }
        } catch (IOException e) {
            notifyState(StateChangeCallback.CLIENT_CLOSE, "Connection error: " + e.getMessage());
            disconnect();
        }
    }
}
